package com.cloud.basic.TcpAndUdp;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.*;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: basic
 * @description: 数据报文接收
 * @author: Cloud
 * @create: 2021/10/21 14:03:19
 */
@Slf4j
public class DatagramReceived {

    // 目标地址
    private SocketAddress remoteAddress;

    private DatagramSocket datagramReceived;

    private SocketAddress localAddress;

    // 本地缓存已发送的数据 Map key 为消息ID, value消息对象本身
    private Map<Integer,ResponseMessage> msgQueue = new ConcurrentHashMap<>();

    public DatagramReceived() throws SocketException {
        localAddress = new InetSocketAddress("127.0.0.1", 14000);
        remoteAddress = new InetSocketAddress("127.0.0.1", 13000);
        datagramReceived = new DatagramSocket(localAddress);

        // 启动接收应答线程
        responseThread();
        // 启动重发线程
        responseThread();
        // 启动接收线程
        receivedThread();
    }

    public void receivedThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    received();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void reSendMsgThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    reSendMsg();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void responseThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    response();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void received() throws IOException {
        System.out.println("开始接收消息！");
        while (true) {
            // 接收发送端发送过来的数据，100表示缓存的长度
            byte[] recData = new byte[100];
            DatagramPacket receivePacket = new DatagramPacket(recData, recData.length);
            datagramReceived.receive(receivePacket);
            // 获取接收端发送的数据
            RequestMessage requestMessage = new RequestMessage(receivePacket.getData());
            String requestData = new String(requestMessage.getData());
            System.out.println("接收到发送端的数据为："+requestData);

            // 将接收到的数据发送给发送端
            byte[] bytes = (requestData + "world").getBytes();
            ResponseMessage responseMessage = new ResponseMessage(requestMessage.getId(), bytes, 0);
            System.out.println("接收返回的数据：" + new String(responseMessage.getData()));
            byte[] bytes1 = responseMessage.toByte();
            DatagramPacket datagramPacket = new DatagramPacket(bytes1, bytes1.length, remoteAddress);
            datagramReceived.send(datagramPacket);

            // 将接收端返回的数据写入到队列中，用于后面监听重发机制
            responseMessage.setLastSendTime(System.currentTimeMillis());
            responseMessage.setSendCount(1);
            responseMessage.setData(responseMessage.getData());
            // 对于接收端来说，它需要返回的地址是请求消息的本地
            responseMessage.setRemoteAddr(remoteAddress);
            msgQueue.put(requestMessage.getId(),responseMessage);

            System.out.println("接收端已应答：" + responseMessage);
        }
    }

    public void reSendMsg() throws IOException {
        Set<Integer> keySet = msgQueue.keySet();
        Iterator<Integer> iterator = keySet.iterator();
        while (iterator.hasNext()) {
            ResponseMessage responseMessage = msgQueue.get(iterator.next());

            // 重发三次就移除
            if (responseMessage.getSendCount() >= 3) {
                iterator.remove();
                System.out.println("发送端--检测道丢失的消息：" + responseMessage);
            }

            long startTime =  System.currentTimeMillis();
            // 等待时间不超过3s
            if (startTime-responseMessage.getLastSendTime() < 3000 && responseMessage.getSendCount() < 3) {
                byte[] buffer = responseMessage.toByte();
                DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length, responseMessage.getRemoteAddr());
                datagramReceived.send(datagramPacket);
                responseMessage.setSendCount(responseMessage.getSendCount()+1);
                System.out.println("客户端重新发送消息:"+responseMessage);
            }
        }
    }

    public void response() throws IOException {
        System.out.println("发送端-接受应答线程启动");
        while (true) {
            byte[] bytes = new byte[100];
            // 创建接收数据包对象
            DatagramPacket packet = new DatagramPacket(bytes, bytes.length);
            datagramReceived.receive(packet);
            RequestMessage requestMessage = new RequestMessage(packet.getData());
            ResponseMessage responseMessage = msgQueue.get(requestMessage.getId());
            if (responseMessage != null) {
                System.out.println("接收端发送的源数据："+responseMessage);
                System.out.println("接收端已收到发送端返回的数据："+ new String(requestMessage.getData()));
                msgQueue.remove(requestMessage.getId());
            }
        }
    }

    public static void main(String[] args) throws SocketException {
        new DatagramReceived();
    }
}
